package cn.wisenergy.pai.monitor.collector;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import cn.wisenergy.pai.monitor.Const;
import cn.wisenergy.pai.monitor.utils.ConfigurationTools;
import cn.wisenergy.pai.monitor.utils.KafkaUtils;

public class ClusterMaster {

	private ConsumerConnector consumer;
	private String topic;
	private ExecutorService executor;
	
	public ClusterMaster() {}
	
	public void shutdown() {
		if (consumer != null)
			consumer.shutdown();
		if (executor != null)
			executor.shutdown();
	}

	public void run(int threadNum) {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(threadNum));
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
		executor = Executors.newFixedThreadPool(threadNum);
		int threadNumber = 0;
		for (final KafkaStream<byte[], byte[]> stream : streams) {
			TimeNodeProcessor processor=new HBaseDataProcessor();
			processor.init(stream, threadNumber);
			executor.submit(processor);
			threadNumber++;
		}
	}

	public ConsumerConnector getConsumer() {
		return consumer;
	}
	public void setConsumer(ConsumerConnector consumer) {
		this.consumer = consumer;
	}
	public String getTopic() {
		return topic;
	}
	public void setTopic(String topic) {
		this.topic = topic;
	}
	public ExecutorService getExecutor() {
		return executor;
	}
	public void setExecutor(ExecutorService executor) {
		this.executor = executor;
	}
	/**
	 * @param args
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException {
		String groupId = args[1];
		String masterIp=ConfigurationTools.getPAIConfValue("JOB_SERVER_ADDR");
		String zookeeper = masterIp + ":2181";
		ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(KafkaUtils.createConsumerConfig(zookeeper, groupId));
		String topic = Const.TOPIC;
		int threads = 1;
		ClusterMaster master = new ClusterMaster();
		master.setConsumer(consumer);
		master.setTopic(topic);
		master.run(threads);
		master.shutdown();

	}

}
